package com.surfilter.massdata.spark.input;

import java.lang.reflect.Field;
import java.util.HashMap;
import java.util.Map;

import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.mapred.InvalidInputException;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.SQLContext;

import com.act.sparkanalyz.input.ISparkInput;
import com.act.sparkanalyz.log.SysLog;
import com.act.sparkanalyz.service.impl.SparkService.ServiceContext;
import com.surfilter.massdata.spark.util.ConfigUtil;

/**
 * spark读取文件输入类
 * 将文件导成dataframe
 * 
 * @author johnguo
 *
 */
public class FileSparkInputforSmcs implements ISparkInput {

	protected static final long serialVersionUID = 1L;
	private String path;
	private String beanName;
	private String tableName;
	protected String splitChar;

	public Map<String, DataFrame> read(ServiceContext serviceContext, Map<String, String> commandMap) throws Exception {
		SQLContext sqlCtx = serviceContext.getSqlCtx();
		String filePath="";
		String exectime="";
		
		if(commandMap.containsKey("-d")){
			exectime= commandMap.get("-d");
			String day=exectime.substring(0, exectime.length()-2);
			filePath=path+"/"+day+"/*"+exectime+"*";
		}else{
			return null; 
		}
		JavaRDD<String> ds =null;
		
		try{
			ds = serviceContext.getCtx().textFile(filePath);
		}catch(Exception e){
			e.printStackTrace();
		}
		SysLog.info("读取 " + filePath + " 文件");
		JavaRDD<Object> tableRDD = ds.map(new Function<String, Object>() {
			private static final long serialVersionUID = 1L;

			@Override
			public Object call(String str) throws Exception {
				String [] line=StringUtils.split(str, splitChar);
				Class<?> beanClass = Class.forName(beanName);
				Object obj = beanClass.newInstance();
				Field[] fields = beanClass.getDeclaredFields();
				Object value = null;
				// SysLog.info("lines==================:" +
				// Arrays.toString(line));
				// SysLog.info("fields==================:" +
				// Arrays.toString(fields));
				//System.out.println("****************line:"+line.length);
				for (int i = 0; i < line.length; i++) {
					//System.out.println("****************line:"+i+":"+line[i]);
					value = null;
					fields[i].setAccessible(true);
					if (line[i] != null) {
						try {
							//switch (fields[i].getGenericType().getTypeName()) {
							switch (fields[i].getType().getName()) {
							case "int":
							case "java.lang.Integer":
								value = Integer.valueOf(line[i]);
								break;
							case "double":
							case "java.lang.Double":
								value = Double.valueOf(line[i]);
								break;
							case "float":
							case "java.lang.Float":
								value = Float.valueOf(line[i]);
								break;
							case "long":
							case "java.lang.Long":
								value = Long.valueOf(line[i]);
								break;
							case "boolean":
							case "java.lang.Boolean":
								value = Boolean.valueOf(line[i]);
								break;
							case "java.lang.String":
								value = line[i];
								break;
							}
							fields[i].set(obj, value);
						} catch (Exception ex) {
							// ex.printStackTrace();
						}
					}
				}
				return obj;
			}
		});
		System.err.println("分区数==========:"+tableRDD.getNumPartitions());
		DataFrame tableFrame = sqlCtx.createDataFrame(tableRDD, Class.forName(beanName));
//		tableFrame.registerTempTable(tableName);
		Map<String, DataFrame> result = new HashMap<String, DataFrame>();
		SysLog.info("*********************FileSparkInputforSmcs*****************************");
		//tableFrame.show();
		result.put(tableName, tableFrame);
		return result;
	}

	protected JavaRDD<String[]> getTableRDD(JavaRDD<String> ds) throws Exception {
		JavaRDD<String[]> td = ds.map(new Function<String, String[]>() {

			private static final long serialVersionUID = 1L;

			@Override
			public String[] call(String line) throws Exception {
				String[] values = line.split(splitChar);
				return values;
			}
		});
		return td;
	}

	protected String getPath(String path) {
		return path;
	}
}
